#include "mainmemorydb.h"
#include "mainconfig.h"
//#include "mainctrl.h"

//是否开启内存数据库
#define MEMORY_BOOT_CHECK(ret);  if(CONF->OpenMemoryDB==0){return ret;}

int CMainMemoryDB::Init(CMainCtrl& stOwern, CTEpoll& stEpoll)
{
    int iRet = 0;
    m_pstOwner = &stOwern;
    m_pstEpoll = &stEpoll;

    iRet = m_stRecvBuffer.Attach(NULL, 65536 * 100, 0);
    CHECK_RETURN(iRet);

    iRet = m_stSendBuffer.Attach(NULL, 65536 * 1000, 0);
    CHECK_RETURN(iRet);

    iRet = m_stEncodeBuff.Attach(NULL, 65536, 0);
    CHECK_RETURN(iRet);

    // 创建socket
    iRet = m_stSocket.Create();
    CHECK_RETURN(iRet);

    // 注册epoll
    iRet = Register(*this, &CMainMemoryDB::OnDBEvent, stEpoll,
        m_stSocket.GetSocket(), EPOLLOUT); // 连接时是监听OUT事件
    CHECK_RETURN(iRet);

    // 连接
    CheckConnect();

    return 0;
}

void CMainMemoryDB::OnDBEvent(CTEpollObject<CMainMemoryDB>* pstObject, SOCKET iSocket, int iEvent)
{
    int iRet = 0;
    CT_TRACE(("recv memdb event %d, socket = %d, connect stat is %d", iEvent, iSocket, m_iConnectState));

    // 连接被断开
    if (CTEpoll::IsCloseEvent(iEvent))
    {
        CT_WARNING(("memdb connect(%s:%hu) close, socket=%d, event = %d", CONF->DBHost(), CONF->DBPort, iSocket, iEvent));
        //m_iConnectState = DB_CLOSED;
        ReConnect();
    }

    // 连接成功
    if (CTEpoll::IsOutputEvent(iEvent))
    {
        CT_TRACE(("memdb connect(%s:%hu) ok, socket=%d", CONF->DBHost(), CONF->DBPort, iSocket));
        m_iConnectState = DB_CONNECTED;
        m_OldConnectFlag = DB_CONNECTED;
        m_ConnectCountFlag = ++m_ConnectCountFlag % 100000000;
        CT_TRACE(("Conncet memdb server Flag=%d", m_ConnectCountFlag));
        // 改为监听IN事件
        iRet = ModRegEvent(EPOLLIN);
        if (iRet)
        {
            CT_ERROR(("ModRegEvent failed %d, errno=%d, socket=%d", iRet, CT_ERRNO, iSocket));
            return;
        }

        // 断线重连后需要发送残留的请求
        CheckSend();
    }

    // 有数据返回(对端断开连接也会出发该事件)
    if (CTEpoll::IsInputEvent(iEvent))
    {
        Recv();
    }
}

void CMainMemoryDB::CheckConnect()
{
    if (m_iConnectState == DB_CLOSED)
    {
        int iRet = 0;
        iRet = m_stSocket.Connect(CONF->MemoryDBHost(), CONF->MemoryDBPort);
        if (iRet)
        {
            CT_WARNING(("memdb connect(%s:%hu) failed, ret=%d", CONF->MemoryDBHost(), CONF->MemoryDBPort, iRet));
            return;
        }

        CT_TRACE(("connect(%s:%hu) ok, socket = %d", CONF->MemoryDBHost(), CONF->MemoryDBPort, m_stSocket.GetSocket()));
        m_iConnectState = DB_CONNECTING;
    }
}


void CMainMemoryDB::Recv()
{
	CT_TRACE(("recv begin"));

    int iRecvLen = 0;
    int iRet = m_stSocket.Recv(m_stRecvBuffer.GetFreeBuf() , m_stRecvBuffer.GetFreeLen() - 1, iRecvLen);
    if (iRet)
    {
        ReConnect();
        if ((m_OldConnectFlag != m_iConnectState))
        {
            CT_TRACE(("disconnection memserver ret = %d Flag=%d state=%d", iRet, m_ConnectCountFlag, m_iConnectState));
            m_OldConnectFlag = m_iConnectState;
        }
    }
    else
    {
        m_stRecvBuffer.Append(iRecvLen);
        m_stRecvBuffer.Clear();
        m_iWaitForCheckSendCount = 0;
    }
    
    CT_TRACE(("recv end"));
}

int CMainMemoryDB::Send(const char* szKey, const char* pszBuf, int iBufLen, int iExestSecs)
{
    if(CONF->OpenMemoryDB==0)
    {
        return 0;
    }

    m_stEncodeBuff.Clear();
    
    int iTotalMsgLen = 0;
    m_stEncodeBuff.Append(sizeof(int));

    sprintf(m_stEncodeBuff.GetFreeBuf(), "set %s 0 %d %d\r\n", szKey, iExestSecs, iBufLen);
    m_stEncodeBuff.Append(strlen(m_stEncodeBuff.GetFreeBuf()));

    if (m_stSendBuffer.GetFreeLen() < (m_stEncodeBuff.GetUsedLen() + iBufLen + 2))
    {
        CT_WARNING(("send buff full"));
        return RET_SYS_ERROR;
    }

    iTotalMsgLen = m_stEncodeBuff.GetUsedLen() - sizeof(int) + iBufLen + 2;
    memcpy(m_stEncodeBuff.GetUsedBuf(), (void*)&iTotalMsgLen, sizeof(int));

    memcpy(m_stSendBuffer.GetFreeBuf(), m_stEncodeBuff.GetUsedBuf(), m_stEncodeBuff.GetUsedLen());
    m_stSendBuffer.Append(m_stEncodeBuff.GetUsedLen());

    memcpy(m_stSendBuffer.GetFreeBuf(), pszBuf, iBufLen);
    m_stSendBuffer.Append(iBufLen);
    m_stSendBuffer.GetFreeBuf()[0] = '\r';
    m_stSendBuffer.GetFreeBuf()[1] = '\n';
    m_stSendBuffer.Append(2);

    if (m_iWaitForCheckSendCount == 0)
    {
        CheckSend();
    }
    
    /*
    int iRet = 0;
    if (m_iConnectState != DB_CONNECTED)
    {
        // 还没连接, 此时尽量缓存
        CT_WARNING(("db's connect state is %d, not connected, store %d bytes",
            m_iConnectState, iBufLen));

        iRet = m_stSendBuffer.Append(pszBuf, iBufLen);
        if (iRet)
        {
            CT_WARNING(("store %d bytes failed, ret=%d, buffer freelen=%d",
                iBufLen, iRet, m_stSendBuffer.GetFreeLen()));
        }

        return iRet;
    }

    int iSendLen = 0;
    iRet = m_stSocket.SendEx(pszBuf, iBufLen, iSendLen);
    if (iRet && iRet != CT_EINPROGRESS && iRet != CT_EWOULDBLOCK)
    {
        CT_WARNING(("socket recv error, ret %d", iRet));
        ReConnect();
        CT_WARNING(("SendEx failed, ret=%d", iRet));
        return iRet;
    }

    if (iSendLen == iBufLen)
    {
        return 0;
    }
    else if (iSendLen < iBufLen)
    {
        // 没发送完, 保存在缓冲区内
        m_stSendBuffer.Append(pszBuf + iSendLen, iBufLen - iSendLen);
        return 0;
    }
    else
    {
        CT_WARNING(("SendEx invalid. %d > %d", iSendLen, iBufLen));
        return -1;
    }
    */

    return 0;
}

void CMainMemoryDB::CheckSend()
{
    if (m_stSendBuffer.GetUsedLen() == 0)
    {
        return;
    }

    if (DB_CONNECTED != m_iConnectState)
    {
        CT_WARNING(("db not connect, now stat is %d, check send return", m_iConnectState));
        return;
    }

    int iRet = 0;
    int iSendLen = 0;

    static int iLeftMsgLen = 0;
    
    if ( iLeftMsgLen <= 0)  //可发送新消息
    {
        if (m_iWaitForCheckSendCount > 0) //上个消息还未返回
        {
            m_iWaitForCheckSendCount--;

            if (m_iWaitForCheckSendCount <= 0)
            {
                CT_WARNING(("wait for mem db return timeout"));
            }

            return;
        }

        memcpy((void*)&iLeftMsgLen, m_stSendBuffer.GetUsedBuf(), sizeof(int));
        m_stSendBuffer.Remove(sizeof(int));
        m_iWaitForCheckSendCount = MEM_DB_MAX_WAIT_FOR_CHECK_SEND_COUNT;
    }

    iRet = m_stSocket.SendEx(m_stSendBuffer.GetUsedBuf(),
        iLeftMsgLen, iSendLen);
    if (iRet && iRet != CT_EINPROGRESS && iRet != CT_EWOULDBLOCK)
    {
        CT_WARNING(("socket send error, ret %d", iRet));
        ReConnect();
        return;
    }

    CT_TRACE(("db need send %d bytes, actually send %d bytes",
        m_stSendBuffer.GetUsedLen(), iSendLen));

    m_stSendBuffer.Remove(iSendLen);
    iLeftMsgLen -= iSendLen;
}

void CMainMemoryDB::ReConnect()
{
    m_iConnectState = DB_CLOSED;

    //重新注册
    Reregister(m_stSocket);

    //重连
    CheckConnect();

    m_iWaitForCheckSendCount = 0;
    m_stSendBuffer.Clear();
}


